<?php
/**
 * @Author: gan
 * @Description:
 * @File:  RedisImpl
 * @Version: 1.0.0
 * @Date: 2021/12/29 9:39 上午
 */

declare(strict_types=1);

namespace RX\Component\MQ\Impl;

use RX\Component\Cache\RedisCache;
use RX\Enums\MQEnum;
use RX\Exception\BizException;

class RedisImpl implements MQInterface
{
    /**
     * MQTool constructor.
     * @param array $config
     */
    public function __construct(array $config = [])
    {
    }


    public function send(array $params)
    {
        $key           = $params["key"] ?? "";
        $message       = $params["message"] ?? "";
        $maxLen        = $params["maxLen"] ?? null;
        $id            = $params["id"] ?? "*";
        $isApproximate = $params["isApproximate"] ?? false;

        if (empty($key) || empty($message)) {
            throw new BizException(MQEnum::class, MQEnum::PARAMS_ERROR);
        }
        return RedisCache::xAdd($key, $message, $maxLen, $id, $isApproximate);
    }


    public function listen(array $params, callable $callback)
    {
        $group    = $params["group"] ?? "";
        $consumer = $params["consumer"] ?? "";
        $stream   = $params["stream"] ?? null;
        $count    = $params["count"] ?? 1;
        $block    = $params["block"] ?? null;

        if (empty($group) || empty($consumer) || empty($stream)) {
            throw new BizException(MQEnum::class, MQEnum::PARAMS_ERROR);
        }
        $result = RedisCache::xReadGroup($group, $consumer, [$stream => ">"], $count, $block);
        return call_user_func($callback, $result);
    }

    /**
     * Redis stream 初始化
     */
    public function init(array $params)
    {
        $stream        = $params["stream"] ?? "";
        $group         = $params["group"] ?? "";
        $message       = $params["message"] ?? "";
        $maxLen        = $params["maxLen"] ?? null;
        $id            = $params["id"] ?? "*";
        $isApproximate = $params["isApproximate"] ?? false;

        if (empty($group) || empty($message) || empty($stream)) {
            throw new BizException(MQEnum::class, MQEnum::PARAMS_ERROR);
        }
        $existResult = RedisCache::exists($stream);
        if (!$existResult) {
            RedisCache::xAdd($stream, $message, $maxLen, $id, $isApproximate);
        }
        $groupInfo = RedisCache::xInfo('GROUPS', $stream);
        if (!empty($groupInfo)) {
            foreach ($groupInfo as $key => $value) {
                if ($value['name'] == $group) {
                    return null;
                }
            }
        }
        return RedisCache::xGroup("CREATE", $stream, $group);
    }


    public function close()
    {
        return true;
    }
}
